package cn.com.bluemoon.bd.flink.source;

import cn.com.bluemoon.bd.flink.pojo.HourMinute;
import cn.com.bluemoon.bd.utils.DateUtils;
import lombok.Data;
import org.apache.commons.lang.math.NumberUtils;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Date;
import java.util.Set;

/**
 * @author 王仁线
 * @version 1.0
 * 用于在每天指定的固定时间点生成消息，主要用于基础数据的加载，其中时间点格式为hh:mm
 */
@Data
public class FixedTimeDataSource extends RichSourceFunction<String> {
    private Set<HourMinute> timePoints;

    public FixedTimeDataSource(String... hourMinutes) {
        this.timePoints = HourMinute.parse(hourMinutes);
    }

    @Override
    public void run(SourceContext ctx) throws Exception {
        while (true) {
            String curHourMinute = DateUtils.dateFormat(new Date(), "HHmm");
            int hour = NumberUtils.toInt(curHourMinute.substring(0, 2));
            int minute = NumberUtils.toInt(curHourMinute.substring(2, 4));
            boolean exists = false;

            for (HourMinute hourMinute : this.timePoints) {
                if (hourMinute.equalTimePoint(hour, minute)) {
                    exists = true;
                    break;
                }
            }

            if (exists) {
                ctx.collect(LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
                Thread.sleep(60 * 1000);
            } else {
                Thread.sleep(30 * 1000);
            }
        }
    }

    @Override
    public void cancel() {

    }

}
